package com.atguigu.flink.datastreamapi.multistream;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.TxEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * Created by Smexy on 2023/4/6
 *
 *  目的： 输出对账成功的订单
 *          格式:  xxxx 订单对账成功。
 *  1.读数据
 *
 *  2.计算
 *
 *  3.输出
 */
public class Demo3_ConnectExec
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读数据
        FileSource<String> fs1 = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("data/OrderLog.csv"))
                                             .build();

        FileSource<String> fs2 = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("data/ReceiptLog.csv"))
                                           .build();

        SingleOutputStreamOperator<OrderEvent> orderEventDs = env
            .fromSource(fs1, WatermarkStrategy.noWatermarks(), "order")
            .map(new MapFunction<String, OrderEvent>()
            {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new OrderEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    );
                }
            })
            .filter(o -> "pay".equals(o.getEventType()));

        SingleOutputStreamOperator<TxEvent> txEventDs = env
            .fromSource(fs2, WatermarkStrategy.noWatermarks(), "pay")
            .map(new MapFunction<String, TxEvent>()
            {
                @Override
                public TxEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new TxEvent(
                        words[0],
                        words[1],
                        Long.valueOf(words[2])
                    );
                }
            });

        //保证同一笔txId的OrderEvent和TxEvent需要发送到下游的同一个Task
        orderEventDs.connect(txEventDs)
                    .keyBy(OrderEvent::getTxId,TxEvent::getTxId)
                    .process(new CoProcessFunction<OrderEvent, TxEvent, String>()
                    {
                        /*
                        1.要点一:
                            OrderEvent value 和 TxEvent value 都是局部变量，只在当前方法的范围内可以被访问到。
                            因此  OrderEvent value 是无法直接访问到  TxEvent value 。
                            需要借助 成员变量(属性) 来进行访问

                         2.要点二:
                            需要准备两个缓存，用于保存先到的OrderEvent或TxEvent。
                            过程：  哪种数据先到，就把自己写入缓存。
                                   后到的其他类型的数据，到缓存中读取，进行对账

                                   缓存的数据:
                                        OrderEvent orderEvent:
                                            txId--->OrderId
                                            Map
                                        TxEvent txEvent:
                                            txId ,
                                            Set
                         */

                        //缓存OrderEvent
                        private Map<String,Long> orderEventCache = new HashMap<>();
                        private Set<String> txEventCache = new HashSet<>();

                        @Override
                        public void processElement1(OrderEvent orderEvent, Context ctx, Collector<String> out) throws Exception {

                            //先到 txEventCache 看要对账的那个 TxEvent 是否已经存在。已经存在，就可以对账。 不存在，把自己写入 orderEventCache
                            String txId = orderEvent.getTxId();
                            Long orderId = orderEvent.getOrderId();

                            if (txEventCache.contains(txId)){
                                out.collect(orderId + " 对账成功...");
                            }else {
                                orderEventCache.put(txId,orderId);
                            }

                        }

                        @Override
                        public void processElement2(TxEvent txEvent, Context ctx, Collector<String> out) throws Exception {
                            //到 orderEventCache 看要对账的那个 OrderEvent是否已经存在，存在就对账成功。 否则，把自己写入txEventCache
                            String txId = txEvent.getTxId();

                            Long orderId = orderEventCache.get(txId);

                            if (orderId != null){
                                out.collect(orderId + " 对账成功...");
                            }else{
                                txEventCache.add(txId);
                            }
                        }
                    })
                    .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
